package com.biconn.bigdata.spider.kafka.listenner;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.biconn.bigdata.common.util.StringUtils;



@Component
public class SpiderListenner{

	/*
	 * 消费医院信息
	*/
	@KafkaListener(topics = {"kafka.scrapy.question.and.answer"})
	public void processMessage(ConsumerRecord<?, ?> record) {
        System.out.println(record.toString());
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        String message = null;
        try {
        	if (kafkaMessage.isPresent()) {
            	message = StringUtils.unicodeToString((String)kafkaMessage.get()).replaceAll("\\\\", "");
    			System.out.println("this is the testTopic send message:" + message);
                
            }
		} catch (Exception e) {
			//异常跳出 do nothing
			System.out.println("this is the testTopic send message:" + message);
		}
        
	}
	
}
